package com.atguigu.edu.realtime220815.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.common.EduConfig;
import com.atguigu.edu.realtime220815.util.PhoenixUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * @Classname DimSinkFunction
 * @Description TODO
 * @Date 2023/2/15 20:13
 * @Created by lzx
 */
public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //主要就是根据第八步过滤下来的数据来写入对应的phoenix中的表
        //这里也有两种情况:第一种就是如果type字段是delete
        String type = value.getString("type");
        //准备拼接sql语句
        StringBuilder sql = new StringBuilder();
        String sinkTable = value.getString("sink_table");
        //拼接之前需要把表中不存在的俩字段(sink_table和type去掉)
        value.remove("sink_table");
        value.remove("type");
        if("delete".equals(type)){
            /*
            delete from 表空间.表名 where 主键 = 多少
             */
            sql.append("delete from "+ EduConfig.PHOENIX_NAMESPACE+"."+sinkTable +" where id = '"+value.getString("id")+"'");
        }else {
            /*
            第二种是其他的,包括insert update 这两种通用的sql为upsert
            upsert into 表空间.表名(所有的字段) values(所有的字段对应的值)
             */
            sql.append("upsert into "+EduConfig.PHOENIX_NAMESPACE+"."+sinkTable+"(");
            StringBuilder values = new StringBuilder();
            Set<Map.Entry<String, Object>> entries = value.entrySet();
            Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, Object> next = iterator.next();
                sql.append(next.getKey()+",");
                values.append("'"+next.getValue()+"',");
            }
            sql.deleteCharAt(sql.length()-1).append(") values(");
            values.deleteCharAt(values.length()-1);
            sql.append(values.toString()).append(")");

        }
        System.out.println(sql.toString());
        PhoenixUtils.executorSql(sql.toString());
    }
}
